package cd.hadoop.storm;

import backtype.storm.Config;
import backtype.storm.StormSubmitter;
import backtype.storm.generated.AlreadyAliveException;
import backtype.storm.generated.InvalidTopologyException;
import backtype.storm.generated.StormTopology;
import backtype.storm.topology.TopologyBuilder;

/**
 * 组织各个处理组件形成一个完整的处理流程，就是所谓的
 * topology（类似与mapreduce程序中的job）
 * 并且将该topology提交给storm集群去运行，topolopy提交到集群后就将
 * 永远无休止地运行，除非认为或者异常退出
 * @author qiqi
 *
 */
public class TopoMain {
	public static void main(String[] args) throws Exception {
		
		TopologyBuilder builder = new TopologyBuilder();
		
		//将我们的splot组件设置到topology中去
		//spout的bolt的id"randomspout",这个值可以任意（最好起得有意义一点）;spout的并发度是4
		builder.setSpout("randomspout", new RandomWordSpout(),4);
		
		//将大写转换bolt组件设置到topology,并且指定他接收randomspout组件的消息
		//告诉BOLT接收哪一个组件的消息shuffleGrouping（分组策略）：随机分组
		builder.setBolt("upperbolt", new UpperBolt(),4).shuffleGrouping("randomspout");
		
		//将添加后缀的bolt组件设置到topology，并且指定他接收upperbolt组件的消息
		builder.setBolt("suffixbolt", new SuffixBolt(),4).shuffleGrouping("upperbolt");
		
		//用builder来创建一个topology
		StormTopology topology = builder.createTopology();
		
		//配置一些topology在集群中运行时的参数
		Config conf = new Config();
		conf.setNumWorkers(4);
		conf.setDebug(true);
		conf.setNumAckers(0);
		
		//将这个topology提交给storm集群运行
		StormSubmitter.submitTopology("demotopo", conf, topology);
	}
}
